#include "kafkaConsumer.h"

KafkaConsumer::KafkaConsumer(const std::string brokers,string topic, std::string groupid, int32_t partition = 0)
        :m_strBroker(brokers),m_strGroupID(groupid),m_partitionSize(partition)
{
    m_topics.push_back(topic);
}

KafkaConsumer::~KafkaConsumer(){
    stop();
}

bool KafkaConsumer::init(){
    std::string errstr;


    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    if (!conf) {
        std::cerr << "RdKafka create global conf failed" << endl;
        return false;
    }

    //ReblanceCallBack reblancecb;
    //conf->set("rebalance_cb",&reblancecb,errstr);

    /*设置broker list*/
    if (conf->set("metadata.broker.list", m_strBroker, errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << "RdKafka conf set brokerlist failed ::" << errstr.c_str() << endl;
    }

    /*设置consumer group*/
    if (conf->set("group.id", m_strGroupID, errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << "RdKafka conf set group.id failed :" << errstr.c_str() << endl;
    }

    std::string strfetch_num = "10240000";
    /*每次从单个分区中拉取消息的最大尺寸*/
    if (conf->set("max.partition.fetch.bytes", strfetch_num, errstr) != RdKafka::Conf::CONF_OK){
        std::cerr << "RdKafka conf set max.partition failed :" << errstr.c_str() << endl;
    }

    /*创建kafka consumer实例*/ //Create consumer using accumulated global configuration.
    m_consumerPtr = RdKafka::KafkaConsumer::create(conf, errstr);
    if (!m_consumerPtr) {
        std::cerr << "failed to create consumer" << endl;
        exit(-1);
    }

    std::cout << "% Created consumer " << m_consumerPtr->name() << std::endl;
    delete conf;

    RdKafka::ErrorCode err=m_consumerPtr->subscribe(this->m_topics);
    if (err) {
    std::cerr << "Failed to subscribe to " << m_topics.size()
              << " topics: " << RdKafka::err2str(err) << std::endl;
    exit(-1);
    }

    
    return true;
}


void KafkaConsumer::message_consume(RdKafka::Message* message,void* opaque){
     string msg="";
    switch (message->err()) {
        case RdKafka::ERR__TIMED_OUT:
            break;

        case RdKafka::ERR_NO_ERROR:

            /* Real message */
           msg=((char*)message->payload());
            if(msg=="end."){
                m_isRun=false;
                break;
            }

            //回调函数不为空
            if(m_msg_fun){
                //执行回调函数
                m_msg_fun(msg);
                cout<<"kafka收到消息，执行回调函数"<<endl;
            }

            printf("%.*s", static_cast<int>(msg.size()),msg.c_str());
            cout<<"\t[offset]:"<<message->offset()<<endl;
            break;

        case RdKafka::ERR__PARTITION_EOF:
            /* Last message */
            m_isRun=false;
            break;

        case RdKafka::ERR__UNKNOWN_TOPIC:
        case RdKafka::ERR__UNKNOWN_PARTITION:
            std::cerr << "Consume failed: " << message->errstr() << std::endl;
            m_isRun = 0;
            break;

        default:
            /* Errors */
            std::cerr << "Consume failed: " << message->errstr() << std::endl;
            m_isRun = 0;
    }

}



void KafkaConsumer::start(int timeout_ms){
    RdKafka::Message* message=nullptr;
    m_isRun=true;

    while(m_isRun){
        message=m_consumerPtr->consume(timeout_ms);

        message_consume(message,nullptr);
        delete message;
        m_consumerPtr->poll(0);
    }

    m_consumerPtr->close();
    m_consumerPtr->poll(1000);


    if(m_consumerPtr){
        delete m_consumerPtr;
        m_consumerPtr=nullptr;
    }

    //销毁kafka实例
    RdKafka::wait_destroyed(5000);
}

void KafkaConsumer::stop(){
    m_isRun=false;
}

//订阅主题
void KafkaConsumer::subscribe(string topic){
    //this->m_topics.push_back(topic); //不能只有这条代码这样来订阅
    std::lock_guard<mutex> lock(m_lock);//上锁

    m_topics.push_back(topic);
    m_consumerPtr->subscribe(m_topics);
}

//取消某个主题的订阅
bool KafkaConsumer::unsubscribe(string topic){
    //使用遍历的方式在topics数组中查看topic
    

    vector<string>::iterator it=m_topics.begin();
    for(;it!=m_topics.end();++it){
        if(*it==topic){
            //找到了
            std::lock_guard<mutex> lock(m_lock);//找到了才上锁，没找到不会对数组进行修改所以不用上锁

            m_topics.erase(it);
            m_consumerPtr->subscribe(m_topics);
            return true;
        }
    }

    return false;
}